package com.bw.utils;

import com.bw.bean.Customer;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.Timestamp;
import java.util.Random;
import java.util.UUID;

public class CustomerSource implements SourceFunction<Customer> {
    private boolean isRunning = true;
    private final Random random = new Random();
    private final String[] memberLevels = {"普通", "白银", "黄金", "铂金", "钻石"};
    private final String[] addresses = {"北京市", "上海市", "广州市", "深圳市", "杭州市", "南京市"};
    private final String[] genders = {"男", "女"};

    @Override
    public void run(SourceContext<Customer> ctx) throws Exception {
        long customerId = 1L;
        while (isRunning) {
            Customer customer = new Customer();
            customer.setCustomerId(customerId++);
            customer.setName("客户" + customerId);
            customer.setPhone("13" + (100000000 + random.nextInt(900000000)));
            customer.setEmail(UUID.randomUUID().toString().substring(0, 8) + "@example.com");
            customer.setGender(genders[random.nextInt(genders.length)]);
            customer.setAge(18 + random.nextInt(60));
            customer.setAddress(addresses[random.nextInt(addresses.length)]);
            customer.setRegisterTime(new Timestamp(System.currentTimeMillis() - random.nextInt(365 * 24 * 3600 * 1000)));
            customer.setMemberLevel(memberLevels[random.nextInt(memberLevels.length)]);
            customer.setTotalConsumption(random.nextDouble() * 10000);

            ctx.collect(customer);

            // 控制数据生成速度
            Thread.sleep(100);

            // 生成100个客户后停止
            if (customerId > 100) {
                break;
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}